package com.neusoft.di;

import java.util.concurrent.Flow;
import java.util.concurrent.TimeUnit;

/**
 * @program: learn
 * @description:
 * @author: Jingdi.Duan ,duanjingdi@msn.com
 * @create: 2024-07-10 10:10
 **/

public class SimpleOnceRequestDemo {
    public static void main(String[] args) {
        //0.1 创建消息发布者，以消息类型以Integer为例
        System.out.println("Main: 0.1 create a publisher with integer message type");
        Flow.Publisher<Integer> publisher = new Flow.Publisher<Integer>() {
            private int start = 0;
            //假设这个publisher产生[0-size)的消息
            private int size = Integer.MAX_VALUE;
            private boolean isCancelled = false;

            @Override
            public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
                //1.1 创建订阅元Subscription
                System.out.println("Publisher: 1.1 create a subscription to the Subscriber");
                Flow.Subscription subscription = new Flow.Subscription() {
                    @Override
                    public void request(long n) {
                        //2.1 接受到消息请求以及请求数量n
                        System.out.println("Publisher: 2.1 receive the request with size from Subscriber");
                        if (isCancelled) {
                            //如果已经收到了取消请求，若再收到request请求，则通知错误
                            subscriber.onError(new IllegalStateException("the subscription is already cancelled"));
                        }
                        //开始发送请求数量的消息
                        long after = start + n;
                        for (int i = start; i < after; i++) {
                            if (i == size) {
                                //2.3 到了数据源的尽头，发送complete消息
                                System.out.println("Publisher: 2.3 send complete message to the Subscriber");
                                subscriber.onComplete();
                                return;
                            }
                            //2.2 发送消息给订阅者
                            System.out.println("Publisher: 2.2 send the message for request size to Subscriber");
                            subscriber.onNext(i);

                        }
                        start = (int) after;
                    }

                    @Override
                    public void cancel() {
                        System.out.println("Publisher: 2.4 receive the cancel request from Subscriber");
                        isCancelled = false;
                    }
                };
                //1.2 传递订阅元（Subscription）给Subscriber
                System.out.println("Publisher: 1.2 send the subscription to the Subscriber");
                subscriber.onSubscribe(subscription);
            }
        };
        //0.2 创建订阅者
        System.out.println("Main: 0.2 create a subscriber");
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<Integer>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription s) {
                //1.3 收到来自publisher的订阅元subscription
                System.out.println("Subscriber: 1.3 receive the subscription from the Publisher");
                subscription = s;
                //2.0 开始发送消息的请求以及请求数量
                System.out.println("Subscriber: 2.0 send the request with size Long.MAX_VALUE");
//                可以一次请求所有的生产者数据
//                subscription.request(Long.MAX_VALUE);
//                或者多次反复请求
                for (int i = 0; i < 100; i++) {
                    subscription.request(2);
                    try {
//                        模拟业务处理....
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }

            @Override
            public void onNext(Integer msg) {
                System.out.println("\tSubscriber: receive the message: " + msg);
            }

            @Override
            public void onError(Throwable t) {
                System.out.println("get error: " + t);
            }

            @Override
            public void onComplete() {
                System.out.println("\tSubscriber: the publisher is finished to send data");
            }
        };
        //0.3 开始订阅
        System.out.println("Main: 0.3 start to subscribe");
        publisher.subscribe(subscriber);
    }
}

























